package day05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Iterator;


public class TestWaterMark03 {
    public static void main(String[] args) throws Exception {
        OutputTag<String> late = new OutputTag<>("late", BasicTypeInfo.STRING_TYPE_INFO);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//时间特性（语义） 默认 ProcessingTime
        env.getConfig().setAutoWatermarkInterval(1000);//间歇性生成Watermark的时间 默认200ms获取一次
        env.addSource(new SourceFunction<String>() {
            Boolean flag = true;
            int num = 1;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (flag) {//当前时间作为事件时间
                    long processing = System.currentTimeMillis();
//                    if (num % 2 == 0) {
//                        processing -= 4000;
//                    }
                    String line = processing + "\thainiu\t" + num;
                    System.out.println("source:" + line);
                    Thread.sleep(1000);
                    ctx.collectWithTimestamp(line, processing);
                    num++;
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
            @Override//watermark 为eventime - 1ms 认为（eventime - 1ms）及以前的人都来了
            public long extractAscendingTimestamp(String element) {
                return Long.valueOf(element.split("\t")[0]);
            }
        }).keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return value.split("\t")[0];
            }
        }).timeWindow(Time.seconds(10),Time.seconds(15))
                //.window(TumblingEventTimeWindows.of(Time.seconds(10),Time.seconds(5))) //一个参数是滚动窗口 二个参数是偏移量
                .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        System.out.println("subtask:" + getRuntimeContext().getIndexOfThisSubtask() +
                                ",start:" + context.window().getStart() +
                                ",end:" + context.window().getEnd() +
                                ",waterMarks:" + context.currentWatermark() +
                                ",currentTime:" + System.currentTimeMillis());

                        Iterator<String> iterator = elements.iterator();
                        for (; iterator.hasNext(); ) {
                            String next = iterator.next();
                            System.out.println("windows-->" + next);
                            out.collect("on time:" + next);
                        }
                    }
                }).print();





        //AssignerWithPunctuatedWatermarks   每过来一个拉一次
//                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
//                                                                //间隔性的拉一次
//                                                                // 子类
//                                                                    //AscendingTimestampExtractor 升序的 watermark加与不加无所谓
//                                                                    //BoundedOutOfOrdernessTimestampExtractor 一定要加
//                    Long current_timestap=0L;
//                    @Nullable
//                    @Override//间歇性  生成Watermark 默认200ms获取一次
//                    public Watermark getCurrentWatermark() {
//                        //间隔时间获取waterMark'
//                        System.out.println("getCurrentWatermark");
//                        return new Watermark(current_timestap-2000);
//                    }
//
//                    @Override//提取时间戳 来一条提取一次(1s一条)
//                    public long extractTimestamp(String element, long previousElementTimestamp) {
//                        System.out.println("extractTimestamp");
//                        //数据获取时间戳
//                        String[] split = element.split("\t");
//                        Long current_timestap = Long.valueOf(split[2]);
//                        this.current_timestap= current_timestap;
//                        return 0;
//                    }
//                })
//                .print();
        /**
         * 确定性的
         */
//        .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
//            @Nullable
//            @Override
//            public Watermark checkAndGetNextWatermark(String lastElement, long extractedTimestamp) {
//                return new Watermark(extractedTimestamp-2000);
//            }
//
//            @Override
//            public long extractTimestamp(String element, long previousElementTimestamp) {
//                System.out.println("extractTimestamp");
//                String[] split = element.split("\t");
//                Long current_timestap = Long.valueOf(split[2]);
//                return current_timestap;
//            }
//        }).print();


                /**
                 *
                 * 间歇性 升序的
                 */
//                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
//                    @Override
//                    public long extractAscendingTimestamp(String element) {
//                        return Long.valueOf(element.split("\t")[2]);
//                        //watermark 为eventime - 1ms 认为（eventime - 1ms）及以前的人都来了
//                    }
//                }).print();

                //乱序


        env.execute();
    }
}
